package com.zhao.rabbitmq.four;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.zhao.rabbitmq.utils.RabbitMQUtils;

import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author xiaoZhao
 * @date 2022/12/4
 * @describe 单个确认发布
 */
public class ConfirmMessage {

    // 发消息的个数
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        // 1.单个确认
        //publishMessageIndividually();

        // 2.批量发布
        //publishmessageBatch();

        // 3.移步发布
        publishMessageAsync();
    }

    // 单个确认
    public static void publishMessageIndividually() throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        // 开始时间
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            StringBuilder message = new StringBuilder();
            message.append(i);
            message.append(" ");
            channel.basicPublish("", queueName, null, message.toString().getBytes());
            // 单个消息发布确认
            boolean flage = channel.waitForConfirms();
            if (flage) {
                System.out.println("消息发送成功");
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("单个发布确认，发布1000条消息耗时：" + (endTime - startTime) + "ms");
    }

    // 批量发布
    public static void publishmessageBatch() throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        // 开始时间
        long startTime = System.currentTimeMillis();
        // 批量确认消息大小
        int batchSize = 100;

        // 批量发送，批量发布确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes());
            //  到达100条消息确认一次
            if (i % batchSize == 0) {
                System.out.println("消息发送成功");
                //  发布确认
                channel.waitForConfirms();
            }
        }

        long endTime = System.currentTimeMillis();
        System.out.println("批量发布确认，发布1000条消息耗时：" + (endTime - startTime) + "ms");
    }

    // 异步发布
    public static void publishMessageAsync() throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        // 声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        // 开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.将序号与消息进行关联
         * 2.批量删除条目
         * 3.支持高并发
        **/
        ConcurrentSkipListMap<Long,Object> outstandingConfirms = new ConcurrentSkipListMap<>();


        // 开始时间
        long startTime = System.currentTimeMillis();

        /**
         * 消息确认成功回调
         * 1. 消息的标识
         * 2.是否为批量确认
        **/
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple){
                // 2.删除已经确认的消息
                ConcurrentNavigableMap<Long, Object> confirmed = outstandingConfirms.headMap(deliveryTag);
                confirmed.clear();
            }else {
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认消息："+deliveryTag);
        };
        // 消息确认失败回调
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            // 3.打印未确认的消息
            Object message = outstandingConfirms.get(deliveryTag);
            System.out.println("未确认消息为："+message+"标记："+deliveryTag);
        };
        // 消息监听器
        channel.addConfirmListener(ackCallback,nackCallback);

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + " ";
            channel.basicPublish("", queueName, null, message.getBytes());
            // 1.记录要发送的所有消息
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }
        //  发布确认

        long endTime = System.currentTimeMillis();
        System.out.println("异步发布确认，发布1000条消息耗时：" + (endTime - startTime) + "ms");
    }

}
